package org.ddpush.im.v1.node.udpconnector;

import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicLong;

import org.ddpush.im.util.PropertyUtil;
import org.ddpush.im.v1.node.Constant;
import org.ddpush.im.v1.node.ServerMessage;

/**
 * UDP 数据发送类
 *
 */
public class Sender implements Runnable {
	/**
	 * UDP 通道。
	 */
	protected DatagramChannel channel;
	/**
	 * 数据包收包计数器
	 */
	protected AtomicLong queueIn = new AtomicLong(0);
	/**
	 * 数据包 发包计数器
	 */
	protected AtomicLong queueOut = new AtomicLong(0);

	/**
	 * 定义 UDP 发射的数据包的最大缓冲区大小
	 */
	protected int bufferSize = Constant.PUSH_MSG_HEADER_LEN + PropertyUtil.getPropertyInt("PUSH_MSG_MAX_CONTENT_LEN");

	/**
	 * 服务器是否被暂停标志
	 */
	protected boolean stoped = false;
	/**
	 * byte 缓冲
	 */
	protected ByteBuffer buffer;

	/**
	 * 这个就暂且看做是一个线程锁对象吧
	 */
	protected Object enQueSignal = new Object();

	/**
	 * 并发线程安全队列。这里主要用来存放需要下发的消息
	 */
	protected ConcurrentLinkedQueue<ServerMessage> mq = new ConcurrentLinkedQueue<ServerMessage>();

	public Sender(DatagramChannel channel) {
		this.channel = channel;
	}

	/**
	 * 初始化、 给byte缓冲区分配一块存储区域、大小是UDP发射的数据包的最大byte个数
	 * 
	 */
	public void init() {
		buffer = ByteBuffer.allocate(bufferSize);
	}

	/**
	 * 停止
	 */
	public void stop() {
		this.stoped = true;
	}

	/**
	 * UDP 数据发射器的线程体
	 */
	public void run() {
		// 这个发射器在服务器启动时 就开始一直运行。直到服务器停止
		while (!this.stoped) {
			try {
				synchronized (enQueSignal) {
					// 如果没有消息 并且 服务器没有关闭。就一直等待
					while (mq.isEmpty() == true && stoped == false) {
						try {
							enQueSignal.wait(1);
						} catch (InterruptedException e) {

						}
						// System.out.println("sender wake up");
					}

					// 如果 有消息了 或者 服务器被关闭了。就开始处理UDP数据
					// 关于 服务器被关闭了 还处理消息这个可以不用纠结
					// 应为服务器关闭了。就必去去跳出内部的线程锁 执行外部循环
					// 外部循环才是判断服务器是否被关闭。所以得走到这里。
					processMessage();

				}
			} catch (Exception e) {
				e.printStackTrace();
			} catch (Throwable t) {
				t.printStackTrace();
			}
		}
	}

	/**
	 * 处理要发送的UDP数据
	 * 
	 * @throws Exception
	 */
	protected void processMessage() throws Exception {
		// 清空缓冲区。
		buffer.clear();
		// 从并发线程安全的消息队列中取出一个被封装的待发射的UDP消息
		ServerMessage pendingMessage = dequeue();
		// 如果消息为空。就返回
		if (pendingMessage == null) {
			// Thread.yield();
			return;
		}
		// 将待发射的UDP消息写入缓冲区
		buffer.put(pendingMessage.getData());
		buffer.flip();
		// 发射这个UDP数据包
		channel.send(buffer, pendingMessage.getSocketAddress());
		// System.out.println(DateTimeUtil.getCurDateTime()+"
		// s:"+StringUtil.convert(pendingMessage.getData())+" to
		// :"+pendingMessage.getSocketAddress().toString());
	}

	/**
	 * 入队 将一个待发射的UDP数据包放入消息队列 并更新入队的消息个数
	 * 
	 * @param message
	 * @return
	 */
	protected boolean enqueue(ServerMessage message) {
		boolean result = mq.add(message);
		if (result == true) {
			queueIn.addAndGet(1);
		}
		return result;
	}

	/**
	 * 出队 取出一个待发射的UDP数据包 并更新 取出的消息个数
	 * 
	 * @return
	 */
	protected ServerMessage dequeue() {
		ServerMessage m = mq.poll();
		if (m != null) {
			queueOut.addAndGet(1);
		}
		return m;
	}

	/**
	 * 发射一个UDP数据包、这个主要用来在外部调用、外部服务器像客户端发送UDP数据包来调用
	 * 
	 * @param message
	 * @return
	 */
	public boolean send(ServerMessage message) {
		return enqueue(message);
	}
}